package com.flink.window;

import com.flink.datasource.UserSource;
import com.flink.entity.User;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.sql.Timestamp;
import java.time.Duration;

/**
 * 描述:
 * 增量聚合函数
 *
 * @author yanzhengwu
 * @create 2022-07-31 18:57
 */
public class WindowAggregateFunction {
    public static void main(String[] args) throws Exception {

        //声明执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //测试为了保证元素顺序并行度设置为1，可以理解为单线程执行
        env.setParallelism(1);
        //设置水位线生成的间隔 这里给的是100毫秒 ,flink 默认是200毫秒 ，flink 可以达到毫秒级别的效率
        env.getConfig().setAutoWatermarkInterval(100);


        //TODO 无序流的watermark生成策略
        DataStream<User> stream = env.addSource(new UserSource())       //生成水位线和时间戳的策略对象
                .assignTimestampsAndWatermarks(
                        //返回一个具体的策略对象(TODO 这里是乱序流的处理方法，给了一个延迟2秒的策略，也可由理解为 数据延迟多长时间能够全部到位)
                        WatermarkStrategy.<User>forBoundedOutOfOrderness(Duration.ofSeconds(2L))
                                //返回策略对象的具体实现
                                .withTimestampAssigner(new SerializableTimestampAssigner<User>() {
                                    /**
                                     * 此方法为指定以事件时间的具体时间戳字段
                                     * @param element
                                     * @param recordTimestamp
                                     * @return 返回的则是一个毫秒数的时间戳
                                     */
                                    @Override
                                    public long extractTimestamp(User element, long recordTimestamp) {
                                        return element.getTimestamp();
                                    }
                                }));

        stream.keyBy(User::getName)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                //TODO 增量聚合函数，可以更好的返回不通的类型，泛型有三个，输入类型、累加器类型、输出类型 都可以用不同的类型 更加的灵活，比使用map和reduce操作更加方便
                //统计user对象中的时间戳叠加之和，和个数
                .aggregate(new AggregateFunction<User, Tuple2<Long, Integer>, String>() {
                    //创建一个累加器,正常初始化给0就行（创建窗口只执行一次）
                    @Override
                    public Tuple2<Long, Integer> createAccumulator() {
                        return Tuple2.of(0L, 0);
                    }

                    //累加逻辑，每次 来一个数据都会调用一次 add方法
                    @Override
                    public Tuple2<Long, Integer> add(User value, Tuple2<Long, Integer> accumulator) {
                        //累加时间戳和 数量
                        return Tuple2.of(accumulator.f0 + value.getTimestamp(), accumulator.f1 + 1);
                    }

                    //返回指逻辑，只要触发一次计算则执行一次（求 一个时间戳的平均值 ）
                    @Override
                    public String getResult(Tuple2<Long, Integer> accumulator) {
                        //用所有时间戳之和除以总数 则得到一个平均值
                        return new Timestamp(accumulator.f0 / accumulator.f1).toString();
                    }

                    //合并操作一般没有并行流则不写(正常回话窗口则会涉及此类方法)
                    @Override
                    public Tuple2<Long, Integer> merge(Tuple2<Long, Integer> a, Tuple2<Long, Integer> b) {
                        //所谓合并就是将值累加在一起
                        return Tuple2.of(a.f0 + b.f0, a.f1 + b.f1);
                    }
                }).print();


        env.execute();
    }
}
